"""
股票数据提供器

基于AKShare和BaoStock提供股票数据获取接口。
支持技术分析、消息分析、估值分析、资金分析等维度的数据获取。
"""

import asyncio
import pandas as pd
from typing import Dict, Any, Optional, List
from pathlib import Path
import os
import hashlib
from datetime import datetime, timedelta

try:
    import akshare as ak
    AKSHARE_AVAILABLE = True
except ImportError:
    AKSHARE_AVAILABLE = False

try:
    import baostock as bs
    BAOSTOCK_AVAILABLE = True
except ImportError:
    BAOSTOCK_AVAILABLE = False


class StockDataProvider:
    """股票数据提供器"""

    def __init__(self, cache_dir: str = "cache/stock_data", cache_ttl_hours: int = 24):
        self.cache_dir = Path(cache_dir)
        self.cache_ttl_hours = cache_ttl_hours
        self.cache_dir.mkdir(parents=True, exist_ok=True)

        # 检查依赖
        if not AKSHARE_AVAILABLE:
            raise ImportError("AKShare不可用，请安装: pip install akshare")
        if not BAOSTOCK_AVAILABLE:
            raise ImportError("BaoStock不可用，请安装: pip install baostock")

    def _get_cache_key(self, api_name: str, params: Dict[str, Any]) -> str:
        """生成缓存键"""
        key_str = f"{api_name}_{str(sorted(params.items()))}"
        return hashlib.md5(key_str.encode()).hexdigest()

    def _is_cache_valid(self, cache_file: str) -> bool:
        """检查缓存是否有效"""
        if not os.path.exists(cache_file):
            return False

        file_time = datetime.fromtimestamp(os.path.getmtime(cache_file))
        return datetime.now() - file_time < timedelta(hours=self.cache_ttl_hours)

    def _load_from_cache(self, cache_file: Path) -> Optional[pd.DataFrame]:
        """从缓存加载数据"""
        try:
            if self._is_cache_valid(str(cache_file)):
                return pd.read_pickle(cache_file)
        except Exception as e:
            print(f"加载缓存失败: {e}")
        return None

    def _save_to_cache(self, data: pd.DataFrame, cache_file: Path):
        """保存数据到缓存"""
        try:
            data.to_pickle(cache_file)
        except Exception as e:
            print(f"保存缓存失败: {e}")

    async def get_stock_realtime_data(self, stock_code: Optional[str] = None) -> pd.DataFrame:
        """
        获取股票实时数据

        Args:
            stock_code: 股票代码，如 "000001" 或 "600000"

        Returns:
            实时数据DataFrame
        """
        cache_key = self._get_cache_key("realtime", {"code": stock_code})
        cache_file = self.cache_dir / f"{cache_key}.pkl"

        # 尝试从缓存加载
        cached_data = self._load_from_cache(cache_file)
        if cached_data is not None:
            return cached_data

        try:
            # 使用AKShare获取实时数据 - 修正API名称
            if stock_code:
                # 单个股票实时数据 - 使用spot_em替代
                data = ak.stock_zh_a_spot_em()
                # 过滤指定股票
                if not data.empty:
                    data = data[data['代码'] == stock_code]
            else:
                # 全部A股实时数据
                data = ak.stock_zh_a_spot_em()

            # 保存到缓存
            self._save_to_cache(data, cache_file)
            return data

        except Exception as e:
            raise Exception(f"获取实时数据失败: {e}")

    async def get_stock_history_data(self, stock_code: str, start_date: Optional[str] = None,
                                   end_date: Optional[str] = None, period: str = "daily") -> pd.DataFrame:
        """
        获取股票历史数据

        Args:
            stock_code: 股票代码，如 "000001"
            start_date: 开始日期，如 "20230101"
            end_date: 结束日期，如 "20231231"
            period: 周期，"daily"/"weekly"/"monthly"

        Returns:
            历史数据DataFrame
        """
        params = {
            "code": stock_code,
            "start_date": start_date,
            "end_date": end_date,
            "period": period
        }
        cache_key = self._get_cache_key("history", params)
        cache_file = self.cache_dir / f"{cache_key}.pkl"

        # 尝试从缓存加载
        cached_data = self._load_from_cache(cache_file)
        if cached_data is not None:
            return cached_data

        try:
            # 使用AKShare获取历史数据
            data = ak.stock_zh_a_hist(
                symbol=stock_code,
                period=period,
                start_date=start_date or "20230101",
                end_date=end_date or datetime.now().strftime("%Y%m%d"),
                adjust="qfq"  # 前复权
            )

            # 保存到缓存
            self._save_to_cache(data, cache_file)
            return data

        except Exception as e:
            raise Exception(f"获取历史数据失败: {e}")

    async def get_stock_technical_indicators(self, stock_code: str, start_date: Optional[str] = None,
                                           end_date: Optional[str] = None) -> Dict[str, Any]:
        """
        获取股票技术指标数据

        Args:
            stock_code: 股票代码
            start_date: 开始日期
            end_date: 结束日期

        Returns:
            技术指标字典
        """
        # 获取历史数据
        history_data = await self.get_stock_history_data(stock_code, start_date, end_date)

        if history_data.empty:
            return {}

        # 计算技术指标
        indicators = {}

        try:
            # 移动平均线
            indicators['ma5'] = float(history_data['收盘'].rolling(5).mean().iloc[-1])
            indicators['ma10'] = float(history_data['收盘'].rolling(10).mean().iloc[-1])
            indicators['ma20'] = float(history_data['收盘'].rolling(20).mean().iloc[-1])
            indicators['ma30'] = float(history_data['收盘'].rolling(30).mean().iloc[-1])

            # MACD
            exp1 = history_data['收盘'].ewm(span=12).mean()
            exp2 = history_data['收盘'].ewm(span=26).mean()
            macd = exp1 - exp2
            signal = macd.ewm(span=9).mean()
            indicators['macd'] = float(macd.iloc[-1])
            indicators['macd_signal'] = float(signal.iloc[-1])
            indicators['macd_hist'] = float((macd - signal).iloc[-1])

            # RSI - 修复类型错误
            delta = history_data['收盘'].diff()
            # 确保delta是数值类型
            delta_numeric = pd.to_numeric(delta, errors='coerce').fillna(0)
            gain = (delta_numeric.where(delta_numeric > 0, 0.0)).rolling(14).mean()
            loss = (-delta_numeric.where(delta_numeric < 0, 0.0)).rolling(14).mean()
            rs = gain / loss
            indicators['rsi'] = float((100 - (100 / (1 + rs))).iloc[-1])

            # 布林带
            sma = history_data['收盘'].rolling(20).mean()
            std = history_data['收盘'].rolling(20).std()
            indicators['bollinger_upper'] = float((sma + 2 * std).iloc[-1])
            indicators['bollinger_lower'] = float((sma - 2 * std).iloc[-1])

        except Exception as e:
            print(f"计算技术指标失败: {e}")

        return indicators

    async def get_stock_valuation_data(self, stock_code: str) -> Dict[str, Any]:
        """
        获取股票估值数据 - 增强版，支持多个数据源

        Args:
            stock_code: 股票代码

        Returns:
            估值数据字典
        """
        cache_key = self._get_cache_key("valuation", {"code": stock_code})
        cache_file = self.cache_dir / f"{cache_key}.pkl"

        # 尝试从缓存加载
        try:
            if self._is_cache_valid(str(cache_file)):
                df = pd.read_pickle(cache_file)
                if not df.empty:
                    return df.iloc[0].to_dict()
        except:
            pass

        result = {}

        # 1. 尝试从个股信息API获取估值数据
        try:
            stock_valuation = ak.stock_individual_info_em(symbol=stock_code)
            if not stock_valuation.empty:
                # 从个股信息中提取估值相关数据
                for _, row in stock_valuation.iterrows():
                    key = row['item']
                    value = row['value']
                    result[key] = value
                result['data_source'] = 'individual_info'
        except Exception as e:
            print(f"个股估值信息获取失败: {e}")

        # 2. 如果个股信息获取失败，尝试从财务分析指标获取
        if not result:
            try:
                valuation_indicators = ak.stock_financial_analysis_indicator_em()
                if not valuation_indicators.empty:
                    # 修正列名：实际是'SECURITY_CODE'而不是'股票代码'
                    stock_data = valuation_indicators[valuation_indicators['SECURITY_CODE'] == stock_code]
                    if not stock_data.empty:
                        latest_data = stock_data.iloc[0].to_dict()
                        result.update({
                            'pe_ratio': latest_data.get('市盈率', 0),
                            'pb_ratio': latest_data.get('市净率', 0),
                            'peg_ratio': latest_data.get('PEG', 0),
                            'roe': latest_data.get('净利率', 0),
                            'roa': latest_data.get('总资产报酬率', 0),
                            'data_source': 'financial_analysis'
                        })
            except Exception as e:
                print(f"财务分析指标获取失败: {e}")

        # 3. 尝试从杜邦分析获取估值相关数据
        if not result.get('roe'):
            try:
                dupont_analysis = ak.stock_zh_dupont_comparison_em()
                if not dupont_analysis.empty:
                    # 修正列名：实际是'代码'而不是'股票代码'
                    stock_dupont = dupont_analysis[dupont_analysis['代码'] == stock_code]
                    if not stock_dupont.empty:
                        latest_dupont = stock_dupont.iloc[0].to_dict()
                        result.update({
                            'roe_dupont': latest_dupont.get('净利率-3年平均', 0),
                            'leverage_ratio': latest_dupont.get('权益乘数-3年平均', 0),
                            'turnover_ratio': latest_dupont.get('总资产周转率-3年平均', 0),
                            'data_source': result.get('data_source', '') + '_dupont'
                        })
            except Exception as e:
                print(f"杜邦分析数据获取失败: {e}")

        # 4. 计算PE/PB比率（使用个股信息+财务摘要）
        if not result.get('pe_ratio') and '总市值' in result:
            try:
                financial_abstract = ak.stock_financial_abstract_ths(symbol=stock_code)
                if not financial_abstract.empty:
                    latest_abstract = financial_abstract.iloc[-1] if len(financial_abstract) > 0 else financial_abstract.iloc[0]
                    
                    # 从财务摘要提取关键数据用于计算PE/PB
                    eps = latest_abstract.get('基本每股收益', 0)  # 每股收益
                    bvps = latest_abstract.get('每股净资产', 0)  # 每股净资产
                    current_price = result.get('最新', 0)  # 当前股价
                    
                    # 计算PE和PB
                    if eps and float(eps) > 0:
                        pe_ratio = float(current_price) / float(eps)
                        result['pe_ratio'] = pe_ratio
                        result['pe_ratio_ths'] = pe_ratio
                    
                    if bvps and float(bvps) > 0:
                        pb_ratio = float(current_price) / float(bvps)
                        result['pb_ratio'] = pb_ratio
                        result['pb_ratio_ths'] = pb_ratio
                    
                    # 添加其他有用的指标
                    result.update({
                        'eps': float(eps) if eps else 0,
                        'bvps': float(bvps) if bvps else 0,
                        'roe': latest_abstract.get('净资产收益率', 0),
                        'net_margin': latest_abstract.get('销售净利率', 0),
                        'gross_margin': latest_abstract.get('销售毛利率', 0),
                        'data_source': result.get('data_source', '') + '_ths_abstract'
                    })
            except Exception as e:
                print(f"同花顺财务摘要获取失败: {e}")

        # 保存到缓存（如果有数据）
        if result:
            try:
                pd.DataFrame([result]).to_pickle(cache_file)
            except Exception as e:
                print(f"保存估值数据缓存失败: {e}")

        return result

    async def get_stock_fundamental_data(self, stock_code: str) -> Dict[str, Any]:
        """
        获取股票基本面数据 - 增强版，支持多个数据源

        Args:
            stock_code: 股票代码

        Returns:
            基本面数据字典
        """
        cache_key = self._get_cache_key("fundamental", {"code": stock_code})
        cache_file = self.cache_dir / f"{cache_key}.pkl"

        # 尝试从缓存加载
        try:
            if self._is_cache_valid(str(cache_file)):
                df = pd.read_pickle(cache_file)
                if not df.empty:
                    return df.iloc[0].to_dict()
        except:
            pass

        result = {}

        # 1. 尝试从同花顺获取财务摘要数据
        try:
            financial_abstract = ak.stock_financial_abstract_ths(symbol=stock_code)
            if not financial_abstract.empty:
                # 提取关键财务指标
                latest_data = financial_abstract.iloc[-1] if len(financial_abstract) > 0 else financial_abstract.iloc[0]
                result.update({
                    'revenue': latest_data.get('营业收入', 0),  # 营业收入
                    'net_profit': latest_data.get('净利润', 0),  # 净利润
                    'roe': latest_data.get('净利率', 0),  # 净利率
                    'pe_ratio': latest_data.get('市盈率', 0),  # 市盈率
                    'pb_ratio': latest_data.get('市净率', 0),  # 市净率
                    'gross_margin': latest_data.get('毛利率', 0),  # 毛利率
                    'data_source': 'ths_abstract'
                })
        except Exception as e:
            print(f"同花顺财务摘要获取失败: {e}")

        # 2. 如果财务摘要获取失败，尝试获取资产负债表数据
        if not result:
            try:
                balance_sheet = ak.stock_financial_debt_ths(symbol=stock_code)
                if not balance_sheet.empty:
                    latest_data = balance_sheet.iloc[-1] if len(balance_sheet) > 0 else balance_sheet.iloc[0]
                    result.update({
                        'total_assets': latest_data.get('资产总计', 0),  # 总资产
                        'total_liabilities': latest_data.get('负债合计', 0),  # 总负债
                        'shareholder_equity': latest_data.get('股东权益合计', 0),  # 股东权益
                        'current_assets': latest_data.get('流动资产合计', 0),  # 流动资产
                        'current_liabilities': latest_data.get('流动负债合计', 0),  # 流动负债
                        'data_source': 'ths_balance'
                    })
            except Exception as e:
                print(f"同花顺资产负债表获取失败: {e}")

        # 3. 获取利润表数据
        try:
            income_statement = ak.stock_financial_benefit_ths(symbol=stock_code)
            if not income_statement.empty:
                latest_data = income_statement.iloc[-1] if len(income_statement) > 0 else income_statement.iloc[0]
                result.update({
                    'operating_revenue': latest_data.get('营业收入', 0),  # 营业收入
                    'operating_profit': latest_data.get('营业利润', 0),  # 营业利润
                    'net_profit_attr': latest_data.get('净利润', 0),  # 净利润
                    'eps': latest_data.get('基本每股收益', 0),  # 每股收益
                    'data_source': result.get('data_source', '') + '_income'
                })
        except Exception as e:
            print(f"同花顺利润表获取失败: {e}")

        # 4. 获取现金流量表数据
        try:
            cash_flow = ak.stock_financial_cash_ths(symbol=stock_code)
            if not cash_flow.empty:
                latest_data = cash_flow.iloc[-1] if len(cash_flow) > 0 else cash_flow.iloc[0]
                result.update({
                    'net_cash_flow_op': latest_data.get('经营活动产生的现金流量净额', 0),  # 经营现金流
                    'net_cash_flow_inv': latest_data.get('投资活动产生的现金流量净额', 0),  # 投资现金流
                    'net_cash_flow_fin': latest_data.get('筹资活动产生的现金流量净额', 0),  # 筹资现金流
                    'data_source': result.get('data_source', '') + '_cash'
                })
        except Exception as e:
            print(f"同花顺现金流量表获取失败: {e}")

        # 5. 备用数据源：BaoStock
        if not result:
            try:
                import baostock as bs

                # 登录
                lg = bs.login()
                if lg.error_code == '0':
                    # 获取季频估值指标
                    rs = bs.query_stock_basic(code=stock_code)
                    data_list = []
                    while (rs.error_code == '0') & rs.next():
                        data_list.append(rs.get_row_data())

                    # 登出
                    bs.logout()

                    if data_list:
                        # 转换为DataFrame
                        columns = rs.fields
                        df = pd.DataFrame(data_list, columns=columns)
                        result = df.iloc[0].to_dict()
                        result['data_source'] = 'baostock'
            except Exception as e:
                print(f"BaoStock基本面数据获取失败: {e}")

        # 保存到缓存（如果有数据）
        if result:
            try:
                pd.DataFrame([result]).to_pickle(cache_file)
            except Exception as e:
                print(f"保存基本面数据缓存失败: {e}")

        return result

    async def get_stock_news_sentiment(self, stock_code: str, days: int = 7) -> List[Dict[str, Any]]:
        """
        获取股票新闻和情感分析 - 增强版，支持多个数据源

        Args:
            stock_code: 股票代码
            days: 最近几天

        Returns:
            新闻列表
        """
        cache_key = self._get_cache_key("news", {"code": stock_code, "days": days})
        cache_file = self.cache_dir / f"{cache_key}.pkl"

        # 尝试从缓存加载
        try:
            if self._is_cache_valid(str(cache_file)):
                cached_data = pd.read_pickle(cache_file)
                # 确保返回类型匹配
                return [dict(item) for item in cached_data.to_dict('records')]
        except:
            pass

        result = []

        # 1. 尝试从东方财富获取新闻数据
        try:
            news_data = ak.stock_news_em(symbol=stock_code)
            if not news_data.empty:
                # 过滤最近几天的数据
                end_date = datetime.now()
                start_date = end_date - timedelta(days=days)

                news_data['发布时间'] = pd.to_datetime(news_data['发布时间'])
                recent_news = news_data[news_data['发布时间'] >= start_date]

                if not recent_news.empty:
                    result = [dict(item) for item in recent_news.to_dict('records')]
                    # 添加数据源标识
                    for item in result:
                        item['data_source'] = 'eastmoney'
        except Exception as e:
            print(f"东方财富新闻数据获取失败: {e}")

        # 2. 如果东方财富获取失败，尝试获取公司公告
        if not result:
            try:
                # 修正API调用：stock_gsrl_gsdt_em不接受start_date和end_date参数
                announcements = ak.stock_gsrl_gsdt_em()
                if not announcements.empty:
                    # 修正列名：实际是'代码'而不是'股票代码'
                    stock_announcements = announcements[announcements['代码'] == stock_code]
                    if not stock_announcements.empty:
                        # 进一步按时间过滤（最近days天）
                        end_date = datetime.now()
                        start_date = end_date - timedelta(days=days)

                        # 修正时间列名：实际是'交易日'而不是'公告日期'
                        stock_announcements['交易日'] = pd.to_datetime(stock_announcements['交易日'])
                        recent_announcements = stock_announcements[stock_announcements['交易日'] >= start_date]

                        if not recent_announcements.empty:
                            result = [dict(item) for item in recent_announcements.to_dict('records')]
                            # 添加数据源标识
                            for item in result:
                                item['data_source'] = 'announcement'
                                item['标题'] = f"{item.get('事件类型', '公司公告')}：{item.get('具体事项', '')[:50]}..."
                                item['内容'] = item.get('具体事项', '暂无详情')
                                item['发布时间'] = item.get('交易日', datetime.now())
            except Exception as e:
                print(f"公司公告数据获取失败: {e}")

        # 3. 尝试获取研报数据
        if not result:
            try:
                reports = ak.stock_research_report_em(symbol=stock_code)
                if not reports.empty:
                    # 过滤最近几天的数据
                    end_date = datetime.now()
                    start_date = end_date - timedelta(days=days)

                    reports['发布时间'] = pd.to_datetime(reports['发布时间'])
                    recent_reports = reports[reports['发布时间'] >= start_date]

                    if not recent_reports.empty:
                        result = [dict(item) for item in recent_reports.to_dict('records')]
                        # 添加数据源标识
                        for item in result:
                            item['data_source'] = 'research_report'
                            item['标题'] = item.get('标题', '研报')
                            item['内容'] = item.get('摘要', '暂无摘要')
            except Exception as e:
                print(f"研报数据获取失败: {e}")

        # 保存到缓存（如果有数据）
        if result:
            try:
                pd.DataFrame(result).to_pickle(cache_file)
            except Exception as e:
                print(f"保存新闻数据缓存失败: {e}")

        # 确保返回类型匹配 List[Dict[str, Any]]
        # 将Hashable键转换为str键
        return [{str(k): v for k, v in item.items()} for item in result]

    async def get_stock_capital_flow(self, stock_code: str) -> Dict[str, Any]:
        """
        获取股票资金流向数据 - 增强版，支持多个数据源

        Args:
            stock_code: 股票代码

        Returns:
            资金流向数据字典
        """
        cache_key = self._get_cache_key("capital_flow", {"code": stock_code})
        cache_file = self.cache_dir / f"{cache_key}.pkl"

        # 尝试从缓存加载
        try:
            if self._is_cache_valid(str(cache_file)):
                df = pd.read_pickle(cache_file)
                if not df.empty:
                    return df.iloc[0].to_dict()
        except:
            pass

        result = {}

        # 1. 尝试获取个股资金流向数据
        try:
            flow_data = ak.stock_individual_fund_flow(stock=stock_code)
            if not flow_data.empty:
                # 获取最新数据
                latest_data = flow_data.iloc[0].to_dict()
                result.update({
                    'main_inflow': latest_data.get('主力净流入-净额', 0),
                    'main_inflow_ratio': latest_data.get('主力净流入-净占比', 0),
                    'retail_inflow': latest_data.get('散户净流入-净额', 0),
                    'retail_inflow_ratio': latest_data.get('散户净流入-净占比', 0),
                    'date': latest_data.get('日期', ''),
                    'data_source': 'individual_flow'
                })
        except Exception as e:
            print(f"个股资金流向数据获取失败: {e}")

        # 2. 如果个股资金流向获取失败，尝试获取北向资金数据
        if not result:
            try:
                hsgt_data = ak.stock_hsgt_fund_min_em()
                if not hsgt_data.empty:
                    # 获取最新北向资金数据
                    latest_hsgt = hsgt_data.iloc[-1].to_dict()
                    result.update({
                        'north_inflow': latest_hsgt.get('北向资金流入', 0),
                        'north_outflow': latest_hsgt.get('北向资金流出', 0),
                        'north_net_inflow': latest_hsgt.get('北向资金净流入', 0),
                        'south_inflow': latest_hsgt.get('南向资金流入', 0),
                        'south_outflow': latest_hsgt.get('南向资金流出', 0),
                        'south_net_inflow': latest_hsgt.get('南向资金净流入', 0),
                        'date': latest_hsgt.get('日期', ''),
                        'data_source': 'hsgt_flow'
                    })
            except Exception as e:
                print(f"北向资金数据获取失败: {e}")

        # 3. 尝试获取龙虎榜数据 - 使用正确的API
        try:
            # 注意：龙虎榜API可能需要具体日期，这里获取最近交易日的龙虎榜
            from datetime import datetime, timedelta
            yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
            # 使用正确的龙虎榜API
            dragon_tiger = ak.stock_lhb_detail_em(start_date=yesterday, end_date=yesterday)

            if not dragon_tiger.empty:
                # 过滤指定股票的龙虎榜数据 - 修正列名
                stock_dragon_tiger = dragon_tiger[dragon_tiger['代码'] == stock_code]
                if not stock_dragon_tiger.empty:
                    latest_dt = stock_dragon_tiger.iloc[0].to_dict()
                    result.update({
                        'dragon_tiger_amount': latest_dt.get('龙虎榜成交额', 0),
                        'buy_amount': latest_dt.get('龙虎榜买入额', 0),
                        'sell_amount': latest_dt.get('龙虎榜卖出额', 0),
                        'net_buy_amount': latest_dt.get('龙虎榜净买额', 0),
                        'data_source': result.get('data_source', '') + '_dragon_tiger'
                    })
        except Exception as e:
            print(f"龙虎榜数据获取失败: {e}")

        # 4. 尝试获取成交明细数据来分析资金动向
        if not result.get('main_inflow'):
            try:
                # 获取成交明细（最近1000条）
                trade_detail = ak.stock_zh_a_tick_tx_js(symbol=stock_code)
                if not trade_detail.empty:
                    # 分析大单成交（单笔成交额超过10万）
                    large_trades = trade_detail[trade_detail['成交金额'] > 100000]
                    if not large_trades.empty:
                        total_large_amount = large_trades['成交金额'].sum()
                        buy_large_amount = large_trades[large_trades['性质'] == '买盘']['成交金额'].sum()
                        sell_large_amount = large_trades[large_trades['性质'] == '卖盘']['成交金额'].sum()

                        result.update({
                            'large_trade_total': total_large_amount,
                            'large_buy_amount': buy_large_amount,
                            'large_sell_amount': sell_large_amount,
                            'large_net_inflow': buy_large_amount - sell_large_amount,
                            'data_source': result.get('data_source', '') + '_tick_data'
                        })
            except Exception as e:
                print(f"成交明细数据获取失败: {e}")

        # 保存到缓存（如果有数据）
        if result:
            try:
                pd.DataFrame([result]).to_pickle(cache_file)
            except Exception as e:
                print(f"保存资金流向数据缓存失败: {e}")

        return result

    async def get_stock_basic_info(self, stock_code: str) -> Dict[str, Any]:
        """
        获取股票基本信息

        Args:
            stock_code: 股票代码

        Returns:
            基本信息字典
        """
        cache_key = self._get_cache_key("basic_info", {"code": stock_code})
        cache_file = self.cache_dir / f"{cache_key}.pkl"

        # 尝试从缓存加载
        try:
            if self._is_cache_valid(str(cache_file)):
                df = pd.read_pickle(cache_file)
                if not df.empty:
                    return df.iloc[0].to_dict()
        except:
            pass

        try:
            # 使用AKShare获取基本信息
            info_data = ak.stock_individual_info_em(symbol=stock_code)

            if info_data.empty:
                return {}

            result = {}
            for _, row in info_data.iterrows():
                result[row['item']] = row['value']

            # 保存到缓存
            pd.DataFrame([result]).to_pickle(cache_file)

            return result

        except Exception as e:
            print(f"获取基本信息失败: {e}")
            return {}

    def clear_cache(self, older_than_hours: int = 0):
        """
        清空缓存

        Args:
            older_than_hours: 删除多少小时之前的缓存，0表示删除所有
        """
        try:
            now = datetime.now()

            for cache_file in self.cache_dir.glob("*.pkl"):
                if older_than_hours == 0:
                    cache_file.unlink()
                else:
                    file_time = datetime.fromtimestamp(cache_file.stat().st_mtime)
                    if now - file_time > timedelta(hours=older_than_hours):
                        cache_file.unlink()

        except Exception as e:
            print(f"清空缓存失败: {e}")


# 全局实例
stock_data_provider = StockDataProvider()